<?php

/**
 * DB队列, 高并发下不保证顺序
 * Class DBQueue
 */
class DBQueue extends Model
{
    const table_name = 'queue';

    const status_init = 0;
    const status_notify_start = 1;
    const status_notify_end = 2;
    const error = 1;

    /**
     * 创建消息
     * @param string|array $message json
     * @param string $topic 消息类型
     * @throws Exception
     */
    public static function push($message, $topic='')
    {
        if (is_array($message)) {
            $message = json_encode($message);
        }
        self::link(self::table_name)->insert(['message' => $message, 'topic_name' => $topic]);
    }

    //取出消息
    public static function pop($topic='')
    {
        if (!empty($topic)) {
            return self::link(self::table_name)->where(['status' => 0, 'topic_name' => $topic])->order('id')->select()->getOne();
        } else {
            return self::link(self::table_name)->where(['status' => 0])->order('id')->select()->getOne();
        }

    }

    //更新消息状态
    //0:初始化, 1: 开始通知 2: 通知完毕
    public static function updateStatus($id, $status)
    {
        return self::link(self::table_name)->where(['id' => $id])->updateVal(['status' => $status])->update();
    }

    //更新消息状态
    public static function setError($id, $msg)
    {
        return self::link(self::table_name)->where(['id' => $id])->updateVal(['is_error' => 1, 'reason' => $msg])->update();
    }

    //分发消息, 由crontab定时任务调用
    public static function handle($message)
    {
        try {
            $msgBody = json_decode($message['message'], true);
            $msgId = $message['id'];
            $topic = $message['topic_name'];

            $msgBody['time'] = REQUEST_TIME;
            $msgBody['sign'] = Sign::getMd5Sign(SignConfig::$APP_KEY['queue_db'], $msgBody);

            self::updateStatus($msgId, self::status_notify_start); //开始通知

            $notifyUrls = DBQueueConfig::getTopicNotifyUrl($topic);
            foreach ($notifyUrls as $url) {
                FileLog::ini()->info('db_queue 分发消息开始: '.$url);
                ICurl::ini($url)
                    ->setConnectTimeOutMs(100)
                    ->setExecTimeOutMs(100)
                    ->setPostData($msgBody)
                    ->post();
                FileLog::ini()->info('db_queue 分发消息结束: '.$url);
            }

            self::updateStatus($msgId, self::status_notify_end); //通知结束
            return true;
        } catch (Exception $e) {
            self::setError($msgId, $message['reason'].';'.$e->getMessage());
            return false;
        }

    }

}
